package db.nio;

import db.nio.worker.BackWorker;
import db.nio.worker.Worker;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeoutException;

/**
 * 异步客户端
 * Created by daibin on 16-9-15.
 */
public class NioClient<T> extends NioBase<T> {


    /**
     * 连接地址
     */
    private InetSocketAddress address;

    /**
     * 当前会话
     */
    private ChannelContext<T> context;

    /**
     * 读取超时
     */
    private int timeOut = 3000;

    /**
     * 事件触发器
     */
    private SelectionKey selectionKey;

    /**
     * 线程
     */
    private volatile Worker customerWorker;

    /**
     * 构造函数
     *
     * @param poolExecutor 线程池
     * @param framerClass  协议封装器
     * @param encoder      编码器
     * @param decoder      解码器
     * @param processor    处理器
     */
    public NioClient(ThreadPoolExecutor poolExecutor, Class<? extends Framer> framerClass, Encoder<T> encoder, Decoder<T> decoder, Processor<T> processor) {
        super(poolExecutor, framerClass, encoder, decoder, processor);
    }


    /**
     * 绑定服务器地址
     * @param address
     * @return
     */
    @Override
    public NioClient bind(InetSocketAddress address) {
        this.address = address;
        return this;
    }


    /**
     * 启动
     */
    public void startup() throws Exception {

        customerWorker = new BackWorker(poolExecutor);

        //创建监视器
        Selector selector = Selector.open();
        try {
            //打开通道
            SocketChannel channel = SocketChannel.open();
            channel.configureBlocking(false);
            channel.connect(address);

            //绑定监视器到通道
            channel.register(selector, SelectionKey.OP_CONNECT);

            //3秒连接服务器
            int time = 0;
            while (time++ < timeOut) {
                int nKey = selector.select(1);
                if (nKey > 0) {
                    Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                    while (keyIterator.hasNext()) {
                        SelectionKey key = keyIterator.next();
                        keyIterator.remove();
                        if (key.isConnectable()) {
                            channel = (SocketChannel) key.channel();

                            channel.configureBlocking(false);
                            channel.socket().setTcpNoDelay(true);
                            channel.socket().setKeepAlive(true);
                            selectionKey = channel.register(customerWorker.selector, SelectionKey.OP_READ);
                            channel.finishConnect();

                            Framer framer = framerClass.newInstance();
                            framer.setSelectionKey(selectionKey);

                            context = new ChannelContext(framer, encoder, decoder, processor);
                            selectionKey.attach(context);

                            //开启监听
                            customerWorker.start();
                            return;
                        }
                    }
                }
            }
            throw new TimeoutException("连接服务器超时。");
        }finally {
            selector.close();
        }
    }


    /**
     * 写数据
     * @param object
     * @throws Exception
     */
    public void write(T object){
        this.context.write(object);
    }


    /**
     * 设计超时时间
     * @param timeOut
     */
    public void setTimeOut(int timeOut){
        this.timeOut = timeOut;
    }

    /**
     * 写并接收返回数据
     * @param object
     * @param filter
     * @return
     * @throws Exception
     */
    public T writeAndRead(T object,Filter filter) throws Exception{

        //暂停后台读线程
        this.customerWorker.waite = true;

        Selector selector = null;
        SelectionKey tempKey = null;
        try {
            this.customerWorker.selector.wakeup();

            //写数据
            this.context.write(object);


            //注册临时监听器
            SocketChannel channel = (SocketChannel) selectionKey.channel();
            selector = Selector.open();
            tempKey =channel.register(selector,SelectionKey.OP_READ);

            T rtData = null;
            int time = 0;
            //接收读权限
            while (true && selector.isOpen()) {
                int nKey = selector.select(1);
                if(time++ >timeOut)
                    throw new TimeoutException("read time out");

                if (nKey > 0) {
                    time = 0;
                    Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                    while (keys.hasNext()) {
                        SelectionKey key = keys.next();
                        keys.remove();
                        if (key.isReadable()) {
                            List<T> dataList = this.context.read();
                            if (dataList == null)
                                continue;
                            for(T data :dataList){

                                //数据过滤
                                if (filter == null || filter.filter(data)){
                                    rtData = data;
                                    continue;
                                }

                                //内部处理
                                processor.doProcessor(data,context);
                            }

                            //抓到返回数据
                            if(rtData!= null)
                                return rtData;
                        }
                    }
                }
            }
        }finally {
            this.customerWorker.waite = false;
            tempKey.cancel();
            selector.selectNow();
            try {
                selector.close();
            }catch (IOException e){

            }
        }
        return null;
    }

    /**
     * 关闭
     */
    public void shutdown() throws IOException {
            customerWorker.stop = true;
            context.close();
    }

}
